package com.lvmama.java.rhino.etl.core;

import org.apache.spark.streaming.Duration;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;

public abstract class AbstractSparkStreamingTemplate extends AbstractSparkBaseTemplate {

	private static final long serialVersionUID = -2647872552215434151L;
	
	transient private JavaStreamingContext jssc = null;
	
	@Override
	public final void run(String[] args) {
		// 创建Spark Streaming上下文
		jssc = new JavaStreamingContext(sparkConf, new Duration(getIntervalTime()));
		JavaDStream<String>  lines = createJavaDStream(jssc);
		excute(lines);
		jssc.start();
		jssc.awaitTermination();
	}

	/**
	 * 获取Spark Streaming的批量时间
	 * @return
	 */
	public abstract long getIntervalTime();
	
	public abstract void excute(JavaDStream<String> lines);

	public abstract JavaDStream<String> createJavaDStream(JavaStreamingContext jssc);

	@Override
	public void shutdown(String[] args) {
		jssc.close();
		saveEnvironment();
	}
	
	public abstract void saveEnvironment();
}
